package yunjiao.javatutorials.guava.concurrent.transform.assembly;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 数据聚合服务
 *
 * @author yangyunjiao
 */
public class DataAggregationService {
    private final ListeningExecutorService executor =
            MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(15));

    /**
     * 聚合多个数据源的用户信息
     */
    @SuppressWarnings("unchecked")
    public ListenableFuture<UserProfile> getUserProfileAsync(String userId) {
        System.out.println(">>>聚合多个数据源的用户信息");
        // 并行获取多个数据源
        ListenableFuture<UserBasicInfo> basicInfoFuture = getUserBasicInfoAsync(userId);
        ListenableFuture<List<Order>> ordersFuture = getUserOrdersAsync(userId);
        ListenableFuture<List<Payment>> paymentsFuture = getPaymentHistoryAsync(userId);
        ListenableFuture<UserPreferences> preferencesFuture = getUserPreferencesAsync(userId);

        // 等待所有数据源完成
        ListenableFuture<List<Object>> allDataFuture = Futures.allAsList(
                basicInfoFuture, ordersFuture, paymentsFuture, preferencesFuture
        );

        // 聚合数据
        return Futures.transform(
                allDataFuture,
                results -> {
                    UserBasicInfo basicInfo = (UserBasicInfo) results.get(0);
                    List<Order> orders = (List<Order>) results.get(1);
                    List<Payment> payments = (List<Payment>) results.get(2);
                    UserPreferences preferences = (UserPreferences) results.get(3);

                    return aggregateUserProfile(basicInfo, orders, payments, preferences);
                },
                executor
        );
    }

    /**
     * 带超时和降级的用户信息获取
     */
    public ListenableFuture<UserProfile> getUserProfileWithTimeoutAsync(String userId) {
        ListenableFuture<UserProfile> profileFuture = getUserProfileAsync(userId);

        // 添加超时控制
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        ListenableFuture<UserProfile> timeoutFuture = Futures.withTimeout(
                profileFuture, 2, TimeUnit.SECONDS, scheduler
        );

        // 超时或失败时返回基础信息
        return Futures.catching(
                timeoutFuture,
                Exception.class,
                exception -> {
                    System.out.println("获取完整用户档案失败，返回基础信息" + exception.getMessage());
                    return getBasicUserProfile(userId);
                },
                executor
        );
    }

    private ListenableFuture<UserBasicInfo> getUserBasicInfoAsync(String userId) {
        System.out.println("getUserBasicInfoAsync");
        return executor.submit(() -> {
            Thread.sleep(100);
            return new UserBasicInfo(userId, "张三", "zhangsan@example.com");
        });
    }

    private ListenableFuture<List<Order>> getUserOrdersAsync(String userId) {
        System.out.println("getUserOrdersAsync");
        return executor.submit(() -> {
            Thread.sleep(150);
            return Arrays.asList(
                    new Order("order1", 199.0, "COMPLETED"),
                    new Order("order2", 299.0, "PENDING")
            );
        });
    }

    private ListenableFuture<List<Payment>> getPaymentHistoryAsync(String userId) {
        System.out.println("getPaymentHistoryAsync");
        return executor.submit(() -> {
            Thread.sleep(120);
            return Arrays.asList(
                    new Payment("payment1", 199.0, "SUCCESS"),
                    new Payment("payment2", 299.0, "PENDING")
            );
        });
    }

    private ListenableFuture<UserPreferences> getUserPreferencesAsync(String userId) {
        System.out.println("getUserPreferencesAsync");
        return executor.submit(() -> {
            Thread.sleep(80);
            return new UserPreferences("zh-CN", "LIGHT", "NOTIFICATIONS_ON");
        });
    }

    private UserProfile aggregateUserProfile(UserBasicInfo basicInfo, List<Order> orders,
                                             List<Payment> payments, UserPreferences preferences) {
        System.out.println("aggregateUserProfile");
        double totalSpent = orders.stream()
                .filter(order -> "COMPLETED".equals(order.status()))
                .mapToDouble(Order::amount)
                .sum();

        int successfulPayments = (int) payments.stream()
                .filter(payment -> "SUCCESS".equals(payment.status()))
                .count();

        return new UserProfile(
                basicInfo,
                orders,
                payments,
                preferences,
                totalSpent,
                successfulPayments
        );
    }

    private UserProfile getBasicUserProfile(String userId) {
        System.out.println("getBasicUserProfile");
        return new UserProfile(
                new UserBasicInfo(userId, "用户", ""),
                Collections.emptyList(),
                Collections.emptyList(),
                new UserPreferences("zh-CN", "SYSTEM", "NOTIFICATIONS_ON"),
                0.0,
                0
        );
    }


}
